#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from threading import Thread, Lock
from queue import Queue, Empty
from random import randint
import time

msg_q = Queue(maxsize=20)
consumer_q = Queue(maxsize=10)
consumer_status = dict()
lock = Lock()


def producer(n):
    while n > 0:
        msg_q.put(n)
        print("produce 生产:", n)
        time.sleep((randint(1, 3)) / 10)
        n -= 1


def scheduler(timeout=5):
    global consumer_status
    _timeout = timeout
    n = 1
    while True:
        # 定时器置0后关闭scheduler
        if msg_q.qsize() == 0 and _timeout < 0:
            print("{}秒内没有可消费数据，调度器停止".format(timeout))
            break
        # 修改consumer_status时对consumer_status加锁
        lock.acquire()
        # 删除消费者状态表中关闭的消费者后释放锁
        consumer_status = {k: v for k, v in consumer_status.items() if v is True}

        lock.release()
        # 如果消息队列有消息且消费者队列未满
        if not msg_q.empty() and not consumer_q.full():
            c = "consumer{}".format(n)
            print("创建 {}".format(c))
            # consumer状态表中将状态设为True
            lock.acquire()
            consumer_status[c] = True
            lock.release()
            # 为consumer分配一个线程，这里不能join，否则该子线程会阻塞产生下一个子线程
            t = Thread(target=consumer, args=(n,))
            t.start()
            # 将新创建的consumer放入队列
            consumer_q.put(t)
            n += 1
            # 重设scheduler的定时器
            _timeout = timeout
        # 定时器递减
        time.sleep(0.2)
        _timeout -= 0.2


def consumer(m):
    while True:
        name = "consumer{}".format(m)
        time.sleep((randint(5, 9)) / 10)
        try:
            # 无阻塞获取消息
            r = msg_q.get_nowait()
        except Empty:
            print("consumer{} 无数据可消费，停止".format(m))
            # 如果取不到消息则停止消费者并将状态置False
            lock.acquire()
            consumer_status[name] = False
            lock.release()
            break
        print("consumer{} 消费: {}".format(m, r))


if __name__ == '__main__':
    t1 = Thread(target=producer, args=(50,))
    t2 = Thread(target=scheduler)

    threads = [t1, t2]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
